package com.atguigu.flink.datastreamapi.divide;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Created by Smexy on 2023/4/4
 */
public class Demo2_ProcessDivide
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("hadoop102", 8888)
                                                        .map(new WaterSensorMapFunction());


        //OutputTag: 代表数据的类型标记信息，属于元数据。  Types.POJO(WaterSensor.class)  TypeInformation.of(WaterSensor.class)
        OutputTag<WaterSensor> s1 = new OutputTag<>("s1", TypeInformation.of(WaterSensor.class));
        OutputTag<WaterSensor> s2 = new OutputTag<>("s2", TypeInformation.of(WaterSensor.class));
        //按照id进行分流
        //返回的都是主流
        SingleOutputStreamOperator<WaterSensor> ds1 = ds.process(new ProcessFunction<WaterSensor, WaterSensor>()
        {
            // Collector<WaterSensor> out ： 用来做主流输出
            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {

               /* ctx.output(s1, value);
                ctx.output(s2, value);
                out.collect(value);
                out.collect(value);
                out.collect(value);*/
                if ("s1".equals(value.getId())) {

                      /*  将这种类型的数据输出到侧流(支流)
                         output(OutputTag<X> outputTag, X value)
                                OutputTag: 代表数据的类型标记信息，属于元数据。*/

                    ctx.output(s1, value);
                } else if ("s2".equals(value.getId())) {
                    ctx.output(s2, value);
                } else {
                    //主流
                    out.collect(value);
                }

            }
        });

        ds1.print("主流，非s1,s2的传感器");
        SideOutputDataStream<WaterSensor> s1DS = ds1.getSideOutput(s1);
        SideOutputDataStream<WaterSensor> s2DS = ds1.getSideOutput(s2);

        s1DS.printToErr("s1");
        s2DS.printToErr("s2");

        env.execute();


    }
}
